"""
练习案02:对数据计算的rdd成员方法进行使用
"""
import json

from my_utils import spark_util

sc = spark_util.get_spark_context()
# ----------------需求1：计算各个城市销售额排序，从大到小-------------------------------------
# 读取文件得到RDD
file_path = "/Users/dongjian/Documents/学习/python-learn-资料/资料/第15章资料/资料/orders.txt"
file_rdd = sc.textFile(file_path)
# 取出json数据
json_str_rdd = file_rdd.flatMap(lambda x: x.split("|"))
# 将json字符串转换为字典
dict_rdd = json_str_rdd.map(lambda data: json.loads(data))
# 取出字典中的城市和销售额数据
city_money_rdd = dict_rdd.map(lambda x: (x["areaName"], int(x["money"])))
# print(city_money_rdd.collect())
# 按城市进行分组按销售额聚合
city_money_group_rdd = city_money_rdd.reduceByKey(lambda a, b: a + b)
# print(city_money_group_rdd.collect())
# 按销售额聚合结果排序 从大到小
money_sort_rdd = city_money_group_rdd.sortBy(lambda money: money[1], ascending=False, numPartitions=1)
print(f"需求1==各个城市销售额从大到小：{money_sort_rdd.collect()}")

# ----------------需求2：全部城市有哪些商品类别在售卖-------------------------------------
category_rdd = dict_rdd.map(lambda d: d["category"]).distinct()
print(f"需求2==全部城市有这些商品类别在售卖：{category_rdd.collect()}")

# ----------------需求3：北京有哪些商品类别在售卖-------------------------------------

beijing_category_rdd = dict_rdd.filter(lambda d1: d1["areaName"] == "北京").map(lambda d2: d2["category"]).distinct()
print(f"需求3==北京有这些商品类别在售卖：{beijing_category_rdd.collect()}")
